package com.bole.reactivestream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.stream.Stream;

/**
 * @author: 叶子
 * @date: 2022/04/10 19:59
 **/
public class ReactiveProcessor extends SubmissionPublisher<String> implements Flow.Subscriber<String> {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("Processor建立订阅关系");
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.println("Processor接收数据："+item);
        //中间处理
        this.submit(item.toUpperCase());//数据发送给最终订阅者
        //背压实现的核心
        this.subscription.request(10);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("Processor出现异常");
        throwable.printStackTrace();
        this.subscription.cancel();
    }

    @Override
    public void onComplete() {

    }
}